package com.iot.websocket;

import com.alibaba.fastjson.JSON;
import com.iot.common.core.domain.entity.SysDept;
import com.iot.common.core.domain.entity.SysUser;
import com.iot.common.utils.StringUtils;
import com.github.pagehelper.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.websocket.Session;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * websocket管理
 * add by caoning
 */
public class WebSocketManager {
    private static final Logger logger = LoggerFactory.getLogger(WebSocketManager.class);

    /**
     * 设备报警事件订阅session
     */
    private static Map<String/*订阅范围*/, Set<String>/*sessionIds*/> gasEventMsgSubMap = new ConcurrentHashMap();
    private static Map<String/*订阅范围*/, Set<String>/*sessionIds*/> quakeEventMsgSubMap = new ConcurrentHashMap();
    private static Map<String/*订阅范围*/, Set<String>/*sessionIds*/> pressureSensorEventMsgSubMap = new ConcurrentHashMap();
    private static Map<String/*订阅范围*/, Set<String>/*sessionIds*/> beidouEventMsgSubMap = new ConcurrentHashMap();
    /**
     * 燃气管道监测设备电池电压事件订阅session
     */
    private static Map<String/*clientNo*/, Set<String>/*sessionIds*/> gasVbatSubMap = new ConcurrentHashMap();
    /**
     * common订阅
     */
    private static Map<String/*clientNo*/, Map<String, Set<String>/*sessionIds*/>> commonSubMap = new ConcurrentHashMap();
    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private volatile static Map<String/*sessionId*/, Session> sessionMap = new ConcurrentHashMap();
    /**
     * 按照订阅主题对客户端分组
     */
    private static Map<String/*topic*/, Set<String>/*sessionIds*/> subscribeMap = new ConcurrentHashMap();
    /**
     * 终端配置变化订阅
     */
    private static Map<String/*clientNo*/, Set<String>/*sessionIds*/> configChangeSubMap = new ConcurrentHashMap();
    /**
     * 终端配置变化订阅
     */
    private static Map<String/*clientNo*/, Set<String>/*sessionIds*/> deviceStatusSubMap = new ConcurrentHashMap();
    /**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
     */
    public static int onlineCount = 0;

    /**
     * 将websocket客户端放入到map保存
     * @param id
     * @param session
     */
    public static void addSession(String id, Session session){
        WebSocketManager.sessionMap.put(id, session);
    }

    public static void removeSession(String id) {
        WebSocketManager.sessionMap.remove(id);
        // 清理订阅
        for(Map.Entry<String, Set<String>> entry : subscribeMap.entrySet()){
            Set<String> value = entry.getValue();
            value.remove(id);
            if (value.isEmpty()){
                subscribeMap.remove(entry.getKey());
            }
        }
    }

    /**
     * 新增订阅
     * @param msg
     * @param session
     */
    public static void addSubscribe(WebSocketMessage msg, Session session){
        SysUser sysUser = getUser(session);
        SysDept dept = sysUser.getDept();
        String ancestors = dept.getAncestors();
        if(StringUtil.isNotEmpty(msg.getTopic())) {
            if(Topic.GAS_EVENT_MSG.equals(msg.getTopic())){
                // 事件信息
                String key = "0/" + dept.getDeptId();
                if(StringUtils.isNotEmpty(ancestors)){
                    key = ancestors.replace(",", "/");
                }
                Set<String> sessionIds = gasEventMsgSubMap.getOrDefault(key, new HashSet<>());
                sessionIds.add(session.getId());
                gasEventMsgSubMap.put(key, sessionIds);
            }
            if(Topic.QUAKE_EVENT_MSG.equals(msg.getTopic())){
                // 事件信息
                String key = "0/" + dept.getDeptId();
                if(StringUtils.isNotEmpty(ancestors)){
                    key = ancestors.replace(",", "/");
                }
                Set<String> sessionIds = quakeEventMsgSubMap.getOrDefault(key, new HashSet<>());
                sessionIds.add(session.getId());
                quakeEventMsgSubMap.put(key, sessionIds);
            }else if(Topic.PS_EVENT_MSG.equals(msg.getTopic())){
                // 事件信息
                String key = "0/" + dept.getDeptId();
                if(StringUtils.isNotEmpty(ancestors)){
                    key = ancestors.replace(",", "/");
                }
                Set<String> sessionIds = pressureSensorEventMsgSubMap.getOrDefault(key, new HashSet<>());
                sessionIds.add(session.getId());
                pressureSensorEventMsgSubMap.put(key, sessionIds);
            }else if(Topic.BEIDOU_EVENT_MSG.equals(msg.getTopic())){
                // 事件信息
                String key = "0/" + dept.getDeptId();
                if(StringUtils.isNotEmpty(ancestors)){
                    key = ancestors.replace(",", "/") + "/" + dept.getDeptId();
                }
                Set<String> sessionIds = beidouEventMsgSubMap.getOrDefault(key, new HashSet<>());
                sessionIds.add(session.getId());
                beidouEventMsgSubMap.put(key, sessionIds);
            }else if(Topic.GAS_VBAT.equals(msg.getTopic())){
                // 电池电压
                String key = msg.dataString();
                Set<String> sessionIds = gasVbatSubMap.getOrDefault(key, new HashSet<>());
                sessionIds.add(session.getId());
                gasVbatSubMap.put(key, sessionIds);
            }else if(Topic.GAS_GPS.equals(msg.getTopic())){
                // 设备定位信息
                String key = msg.dataString();
                Set<String> sessionIds = configChangeSubMap.getOrDefault(key, new HashSet<>());
                sessionIds.add(session.getId());
                configChangeSubMap.put(key, sessionIds);
            }else{
                Map<String, Set<String>> subMap = commonSubMap.getOrDefault(msg.getTopic(), new HashMap());
                String key = msg.dataString();
                Set<String> sessionIds = subMap.getOrDefault(key, new HashSet());
                sessionIds.add(session.getId());
                subMap.put(key, sessionIds);
                commonSubMap.put(msg.getTopic(), subMap);
            }
        }
    }

    /**
     * 取消订阅
     * @param msg
     * @param session
     */
    public static void removeSubscribe(WebSocketMessage msg, Session session){
//        SecurityUtils.getSysUser();
        SysUser sysUser = getUser(session);

        if(StringUtil.isNotEmpty(msg.getTopic())) {
            if(Topic.QUAKE_EVENT_MSG.equals(msg.getTopic())){
                SysDept dept = sysUser.getDept();
                String key = dept.getAncestors().replace(",", "/");
                Set<String> sessionIds = quakeEventMsgSubMap.getOrDefault(key, new HashSet<>());
                sessionIds.remove(session.getId());
                quakeEventMsgSubMap.put(key, sessionIds);
            }else if(Topic.GAS_EVENT_MSG.equals(msg.getTopic())){
                SysDept dept = sysUser.getDept();
                String key = dept.getAncestors().replace(",", "/");
                Set<String> sessionIds = gasEventMsgSubMap.getOrDefault(key, new HashSet<>());
                sessionIds.remove(session.getId());
                gasEventMsgSubMap.put(key, sessionIds);
            }else if(Topic.PS_EVENT_MSG.equals(msg.getTopic())){
                SysDept dept = sysUser.getDept();
                String key = dept.getAncestors().replace(",", "/");
                Set<String> sessionIds = pressureSensorEventMsgSubMap.getOrDefault(key, new HashSet<>());
                sessionIds.remove(session.getId());
                pressureSensorEventMsgSubMap.put(key, sessionIds);
            }else if(Topic.BEIDOU_EVENT_MSG.equals(msg.getTopic())){
                SysDept dept = sysUser.getDept();
                String key = dept.getAncestors().replace(",", "/");
                Set<String> sessionIds = beidouEventMsgSubMap.getOrDefault(key, new HashSet<>());
                sessionIds.remove(session.getId());
                beidouEventMsgSubMap.put(key, sessionIds);
            }else if(Topic.GAS_GPS.equals(msg.getTopic())){
                String key = msg.dataString();
                Set<String> sessionIds = gasEventMsgSubMap.getOrDefault(key, new HashSet<>());
                sessionIds.remove(session.getId());
                configChangeSubMap.put(key, sessionIds);
            }else if(Topic.GAS_VBAT.equals(msg.getTopic())){
                // 订阅终端配置变化
                String key = msg.dataString();
                Set<String> sessionIds = gasVbatSubMap.getOrDefault(key, new HashSet<>());
                sessionIds.remove(session.getId());
                gasVbatSubMap.put(key, sessionIds);
            }else{
                Map<String, Set<String>> subMap = commonSubMap.get(msg.getTopic());
                if(subMap != null){
                    String key = msg.dataString();
                    Set<String> sessionIds = subMap.get(key);
                    if(sessionIds != null){
                        sessionIds.remove(session.getId());
                        subMap.put(key, sessionIds);
                        commonSubMap.put(msg.getTopic(), subMap);
                    }
                }
            }
        }
    }

    public static Set<String> getGasEventMsgSubSessionIds(String key) {
        Set<String> ids = new HashSet<>();
        for(String subKey : gasEventMsgSubMap.keySet()){
            if(key.startsWith(subKey)){
                ids.addAll(gasEventMsgSubMap.get(subKey));
            }
        }
        return ids;
    }

    public static Set<String> getPressureSensorEventMsgSubSessionIds(String key) {
        Set<String> ids = new HashSet<>();
        for(String subKey : pressureSensorEventMsgSubMap.keySet()){
            if(key.startsWith(subKey)){
                ids.addAll(pressureSensorEventMsgSubMap.get(subKey));
            }
        }
        return ids;
    }

    public static Set<String> getBeidouEventMsgSubSessionIds(String key) {
        Set<String> ids = new HashSet<>();
        for(String subKey : beidouEventMsgSubMap.keySet()){
            if(key.startsWith(subKey)){
                ids.addAll(beidouEventMsgSubMap.get(subKey));
            }
        }
        return ids;
    }

    public static Set<String> geQuakeEventMsgSubSessionIds(String key) {
        Set<String> ids = new HashSet<>();
        for(String subKey : quakeEventMsgSubMap.keySet()){
            if(key.startsWith(subKey)){
                ids.addAll(quakeEventMsgSubMap.get(subKey));
            }
        }
        return ids;
    }

    public static Set<String> getConfigChangeSubSessionIds(String key) {
        Set<String> ids = new HashSet<>();
        for(String subKey : configChangeSubMap.keySet()){
            if(key.equals(subKey)){
                ids.addAll(configChangeSubMap.get(subKey));
            }
        }
        return ids;
    }

    public static Set<String> getDeviceStatusSubSessionIds(String key) {
        Set<String> ids = new HashSet<>();
        for(String subKey : deviceStatusSubMap.keySet()){
            if(key.equals(subKey)){
                ids.addAll(deviceStatusSubMap.get(subKey));
            }
        }
        return ids;
    }

    public static Set<String> getTerminalResultSubSessionIds(String key) {
        Set<String> ids = new HashSet<>();
        for(String subKey : gasVbatSubMap.keySet()){
            if(key.equals(subKey)){
                ids.addAll(gasVbatSubMap.get(subKey));
            }
        }
        return ids;
    }

    public static Set<String> getCommonSubSessionIds(String topic, String key) {
        Set<String> ids = new HashSet<>();
        for(String commonKey : commonSubMap.keySet()){
            if(topic.equals(commonKey)){
                for(String subKey : commonSubMap.getOrDefault(topic, new HashMap<>()).keySet()){
                    if(key.equals(subKey)){
                        ids.addAll(gasVbatSubMap.get(subKey));
                    }
                }
            }
        }
        return ids;
    }

    /**
     * 获取
     * @param topic
     */
    public static Map<String, Set<String>> getSubscribeFilterMap(String topic){
        Map<String, Set<String>> sm = new HashMap<>();
        for(String key : subscribeMap.keySet()){
            if(key.contains(topic)){
                sm.put(key, subscribeMap.get(key));
            }
        }
        return sm;
    }

    /**
     * 当前在线人数
     *
     * @return
     */
    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static Session getSession(String sessionId) {
        return sessionMap.get(sessionId);
    }

    /**
     * 发送信息
     *
     * @param sessionId
     * @param message
     * @throws IOException
     */
    public static void sendMessage(String sessionId, String message) {
        //获取session远程基本连接发送文本消息
        try {
            logger.info("websocket发送消息 sessionId:{}, message:{}", sessionId, message);
            WebSocketManager.getSession(sessionId).getBasicRemote().sendText(message);
        } catch (Exception e) {
            e.printStackTrace();
//            logger.error("WebSocket发送消息异常：{}", e.getMessage());
        }
        //this.session.getAsyncRemote().sendText(message);
    }

    /**
     * 发送信息
     * @param sessionIds
     * @param message
     * @throws IOException
     */
    public static void sendMessage(Set<String> sessionIds, String message) {
        //获取session远程基本连接发送文本消息
        Iterator<String> it = sessionIds.iterator();
        while (it.hasNext()) {
            //群发
            try {
                logger.info("websocket发送消息 sessionIds:{}, message:{}", StringUtils.join(sessionIds.iterator(), ","), message);
//                WebSocketManager.getSession(it.next()).getBasicRemote().sendText(message);
                Session session = WebSocketManager.getSession(it.next());
                if(session != null){
                    session.getBasicRemote().sendText(message);
                }
            } catch (Exception e) {
                logger.error("websocket发送消息失败："+e);
                e.printStackTrace();
            }
        }
    }
    public static void sendMessage(Set<String> sessionIds, Object message) {
        synchronized(sessionMap){
            //获取session远程基本连接发送文本消息
            Iterator<String> it = sessionIds.iterator();
            while (it.hasNext()) {
                //群发
                try {
                    logger.info("websocket发送消息 sessionIds:{}, message:{}", StringUtils.join(sessionIds.iterator(), ","), message);
//                WebSocketManager.getSession(it.next()).getBasicRemote().sendText(message);
                    Session session = WebSocketManager.getSession(it.next());
                    if(session != null){
                        session.getBasicRemote().sendText(JSON.toJSONString(message));
//                        session.getBasicRemote().sendObject(message);
                    }
                } catch (Exception e) {
                    logger.error("websocket发送消息失败："+e);
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 群发
     * @param message
     */
    public static void sendAll(String message) {
        for(Session session : sessionMap.values()){
            //群发
            try {
//                logger.info("websocket群发消息 message:{}", message);
                session.getBasicRemote().sendText(message);
            } catch (IOException e) {
//                logger.error("websocket群发消息失败："+e);
                e.printStackTrace();
            }
        }
    }

    public static SysUser getUser(Session session){
        return (SysUser)session.getUserProperties().getOrDefault("sysUser", new SysUser());
    }
}
